struct ulog_output_buffer {
    int inited;
    pthread_mutex_t lock;
    char*   buffer;
    size_t  size;
    size_t  size_max;
};


typedef enum {
    output_none = 0, 
    output_std, 
    output_path,
    output_tcp,
    output_udp,
}output_type;


struct ulog_output {
    output_type type;
    #define LEN_PATH    100
    char str[LEN_PATH];

    union {
        FILE* fp;
        struct ulog_output_buffer* obuffer;
    }dest;
};


struct ulog_outputs {
    #define NUM_OUTPUT  10
    int num;
    struct ulog_output o[NUM_OUTPUT];
};


#define SIZE_OUTPUT_BUFFER  (100*1024)


int _ustr_is_uri(const char* str, char* addr, size_t size, int* port)
{
    int ret = 0;

    #define LEN_ADDR  1000
    char tmp_addr[LEN_ADDR+24];
    int tmp_port;
    int n = sscanf(str, "%1000[^:]:%d", tmp_addr, &tmp_port);
    if(2 == n) {
        ret = 1;
        if(NULL != addr) {
            _ustrcpy(addr, size, tmp_addr);
        }

        if(NULL != port) {
            *port = tmp_port;
        }
    }
    else {
        ret = 0;
    }

    return ret;
}


int utcp_connect_addr(const char* addr, int port)
{
    int fd = -1;

    if ((fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
        return fd;
    }

    struct sockaddr_in servaddr;
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    servaddr.sin_addr.s_addr = inet_addr(addr);

    if (connect(fd, (struct sockaddr *)&servaddr, sizeof(servaddr)) < 0) {
        close(fd);
        fd = -1;
    }

    return fd;
}


int ufd_detect_writeable(int fd, struct timeval* to)
{
    int ret = 0;

    fd_set wfds;
    int retval;

    FD_ZERO(&wfds);
    FD_SET(fd, &wfds);

    retval = select(fd+1, NULL, &wfds, NULL, to);
    ret = (retval>0)?0:-1; 

    return ret;
}


void* _ulog_tcp_cycle(void* arg)
{
    struct ulog_output* o = arg;

    char addr[LEN_ADDR];
    int port;
    char* str = o->str+strlen("tcp");
    _ustr_is_uri(str, addr, LEN_ADDR, &port);

    int fd = utcp_connect_addr(addr, port);
    if(-1 == fd) {
        return NULL;
    }

    void* buf = malloc(SIZE_OUTPUT_BUFFER) ;
    if(NULL == buf) {
        return NULL;
    }
    size_t size_buf = 0;
    int idx = 0;
    int ret;
    struct timeval to = {0, 100*1000};

    int error_on_write = 0;
    while(!error_on_write) {
        _log("if there's buffer need to be sent.\n");
        pthread_mutex_lock(&o->dest.obuffer->lock);
        if(o->dest.obuffer->size > 0) {
            _log("there's buffer need to be sent.\n");
            memcpy(buf, o->dest.obuffer->buffer, o->dest.obuffer->size);
            size_buf = o->dest.obuffer->size;
            idx = 0;

            o->dest.obuffer->size = 0;
        }
        else {
            _log("there's no buffer need to be sent.\n");
        }
        _log("\n");
        pthread_mutex_unlock(&o->dest.obuffer->lock);

        if(0 == o->dest.obuffer->size) {
            usleep(100*1000);
        }

        while(size_buf > 0) {
            ret = ufd_detect_writeable(fd, &to);
            if(0 == ret) {
                int n = write(fd, buf+idx, size_buf);
                _log("write %d with length : %d\n", fd, n);
                if(n <= 0) {
                    error_on_write = 1;
                    break;
                }

                idx += n;
                size_buf -= n;
            }
        }
    }

    return NULL;
}


int _ulog_create_tcp_runtime(struct ulog_output* o)
{
    int ret = 0;
    char* str = o->str+strlen("tcp");
    if(_ustr_is_uri(str, NULL, 0, NULL)) {
        o->dest.obuffer = 
            malloc(sizeof(struct ulog_output_buffer) + SIZE_OUTPUT_BUFFER);
        if(NULL != o->dest.obuffer) {
            pthread_mutex_init(&o->dest.obuffer->lock, NULL);
            o->dest.obuffer->buffer 
                = (void*)o->dest.obuffer + sizeof(struct ulog_output_buffer);
            o->dest.obuffer->size = 0; 
            o->dest.obuffer->size_max = SIZE_OUTPUT_BUFFER;
            o->dest.obuffer->inited = 0;

            pthread_t pth;
            int ret;
            ret =  pthread_create(&pth, NULL, _ulog_tcp_cycle, o);
            if(0 == ret) {
                pthread_detach(pth);
                o->dest.obuffer->inited = 1;
            }
            else {
                fprintf(stderr, "ulogpro : pthread_create error.\n");
            }
        }
        else {
            fprintf(stderr, "ulogpro : malloc error.\n");
        }
    }

    return ret;
}


int _ulog_parse_outputs(struct ulog_outputs* outputs, const char* str)
{
    int ret = 0;

    outputs->num = 0;

    struct _usstring string[NUM_OUTPUT];
    int n = _ustrsplit(str, ",", string, NUM_OUTPUT);
    int i;
    for(i=0; i<n; i++) {
        struct ulog_output* o = &outputs->o[outputs->num];
        _ustrncpy(o->str, LEN_PATH, string[i].s, string[i].len);
        _ustrtrim(o->str, " ", 1, 1, 1);

        if(0 == strlen(o->str) || 0 == strcmp(o->str, "/dev/null")) {
            continue;
        }
        
        outputs->num ++;

        o->type = output_none;
        o->dest.fp = NULL;

        if(0 == strcmp(o->str, "stdout")) {
            o->type = output_std;
            o->dest.fp = stdout;
        }
        else if(0 == strcmp(o->str, "stderr")) {
            o->type = output_std;
            o->dest.fp = stderr;
        }
        else if(0 == strncmp(o->str, "tcp", strlen("tcp"))) {
            o->type = output_tcp;
            _ulog_create_tcp_runtime(o);
        }
        else if(0 == strncmp(o->str, "udp", strlen("udp"))) {
            o->type = output_udp;

        }
        else {
            o->type = output_path;
            o->dest.fp = fopen(o->str, "w");
            if(NULL == o->dest.fp) {
                fprintf(stderr, "ulogproo utputs to <%s> error.\n", o->str);
            }
        }

        if((output_std == o->type || output_path == o->type ) && (NULL != o->dest.fp)) {
            setbuf(o->dest.fp, NULL);
        }
    }

    return ret;
}
